package cn.tedu.hbase

import java.util

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase._
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes

/**
 * HBase基本操作测试类
 */
object TestDriver {
  def main(args: Array[String]): Unit = {
    // 创建配置项,设置zookeeper的参数
    val conf: Configuration = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "mycentos8")
    conf.set("hbase.zookeeper.property.clientPort", "2181")

    // 创建新表
    CreateTableNew(conf, "Test2", "info", "grade")
    // 删除表
    DelectTableNew(conf, "Test2")

    // 创建新表Test1, 如果已经有了则不要再创建了
    CreateTableNew(conf, "Test1", "info", "grade")
    // 添加数据
    InsertTable(conf, "Test1", "wwr", "grade", "age", "23")
    InsertTable(conf, "Test1", "wwr", "grade", "name", "xujie")
    InsertTable(conf, "Test1", "wwr", "grade", "class", "38")
    // 添加新的列簇
    InsertColumnFaimly(conf, "Test1", "High_Speed")
    // 查询所有的表信息
    ScanValue(conf, "Test1")
    // 查找指定row的信息
    getRowKeyValue(conf, "Test1", "wwr")
    // 删除指定row的信息
    deleteRows(conf, "Test1", "wwr")
  }

  /**
   * 添加表
   *
   * @param conf
   * @param tablename
   * @param columnFamily
   */
  def CreateTableNew(conf: Configuration, tablename: String, columnFamily: String*) = {
    // 创建连接
    val conn: Connection = ConnectionFactory.createConnection(conf)
    // 创建'库'操作 admin对象
    val admin = conn.getAdmin.asInstanceOf[HBaseAdmin]
    // 将字符串转化为表名
    val tableName = TableName.valueOf(tablename)
    // TableDescriptorBuilder 表描述生成器
    val table: TableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName)
    for (cf <- columnFamily) {
      // ColumnFamilyDescriptorBuilder 列簇描述生成器
      val info = ColumnFamilyDescriptorBuilder.of(cf)
      // 添加列簇
      table.setColumnFamily(info)
    }
    // 构建表描述
    val descriptor: TableDescriptor = table.build()
    // 创建表
    admin.createTable(descriptor)
    // 关闭接口
    admin.close()
    conn.close()
  }

  /**
   * 增加列簇
   *
   * @param conf
   * @param tableName
   * @param ColumnFamily
   */
  def InsertColumnFaimly(conf: Configuration, tableName: String, ColumnFamily: String*): Unit = {
    // 创建连接
    val conn = ConnectionFactory.createConnection(conf)
    // 创建'库'操作 admin对象
    val admin: HBaseAdmin = conn.getAdmin.asInstanceOf[HBaseAdmin]
    // 遍历插入要插入的列簇##(表)
    for (cf <- ColumnFamily) {
      // 通过.of方法,转化列簇名
      val cff = ColumnFamilyDescriptorBuilder.of(cf)
      admin.addColumnFamily(TableName.valueOf(tableName), cff)
    }
    admin.close()
    conn.close()
  }

  /**
   * 删除表
   *
   * @param conf
   * @param tablename
   */
  def DelectTableNew(conf: Configuration, tablename: String) = {
    // 创建连接
    val conn: Connection = ConnectionFactory.createConnection(conf)
    // 创建'库'操作 admin对象
    val admin = conn.getAdmin.asInstanceOf[HBaseAdmin]
    // 将String类型字符串,修饰为表名
    val tableName = TableName.valueOf(tablename)
    // 使表下架
    admin.disableTable(tableName)
    // 删除表
    admin.deleteTable(tableName)
    // 关闭接口
    admin.close()
    conn.close()
  }

  /**
   * 插入数据
   *
   * @param conf
   * @param tableName
   * @param rowkey
   * @param columnFamily
   * @param column
   * @param value
   */
  def InsertTable(conf: Configuration, tableName: String, rowkey: String, columnFamily: String, column: String, value: String) = {
    // 创建连接
    val conn: Connection = ConnectionFactory.createConnection(conf)
    val name = TableName.valueOf(tableName)
    // 创建表连接对象
    val table: Table = conn.getTable(name)
    // 创建put对象,传递rowKey
    val put: Put = new Put(Bytes.toBytes(rowkey))
    // 添加属性
    put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value))
    // 添加数据到表
    table.put(put)
    table.close()
    conn.close()
  }

  /**
   * 查询表的所有信息
   *
   * @param conf
   * @param tableName
   */
  def ScanValue(conf: Configuration, tableName: String) = {
    val conn = ConnectionFactory.createConnection(conf)
    val tablename = TableName.valueOf(tableName)
    // 获取表
    val table = conn.getTable(tablename)
    // 创建扫描 对象
    val scan = new Scan()
    // 获取扫描结果
    val scanner: ResultScanner = table.getScanner(scan)

    //取该行的cell迭代器
    /* val scanner: CellScanner = result.cellScanner()
    // 迭代这一行的cell
       while(scanner.advance()){
         val cell = scanner.current()
         println(s"RowKey:${Bytes.toString(CellUtil.cloneRow(cell))}")
         println(s"ColumFamily:${Bytes.toString(CellUtil.cloneFamily(cell))}")
         println(s"Qualifier:${Bytes.toString(CellUtil.cloneQualifier(cell))}")
         println(s"Value:${Bytes.toString(CellUtil.cloneValue(cell))}")
         println("==================================================================")
       }
     */

    val cells = scanner.next().rawCells()
    for (cell <- cells) {
      println(s"RowKey:${Bytes.toString(CellUtil.cloneRow(cell))}")
      println(s"ColumFamily:${Bytes.toString(CellUtil.cloneFamily(cell))}")
      println(s"Qualifier:${Bytes.toString(CellUtil.cloneQualifier(cell))}")
      println(s"Value:${Bytes.toString(CellUtil.cloneValue(cell))}")
      println("==================================================================")
    }

    table.close()
    conn.close()
  }

  /**
   * 根据指定rowKey得到相应的
   *
   * @param conf
   * @param tableName
   * @param RowKey
   */
  def getRowKeyValue(conf: Configuration, tableName: String, RowKey: String) = {
    val conn = ConnectionFactory.createConnection(conf)
    val tablename: TableName = TableName.valueOf(tableName)
    val table: Table = conn.getTable(tablename)
    // 通过get对象查询数据
    val get: Get = new Get(Bytes.toBytes(RowKey))
    val result: Result = table.get(get)
    // 查找指定表,指定属性的值
    println(s"Value:${Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")))}")
    val cells = result.rawCells()
    for (cell <- cells) {
      println(s"Row:${result.getRow}")
      println(s"Row:${Bytes.toString(CellUtil.cloneRow(cell))}")
      println(s"ColumFamily:${Bytes.toString(CellUtil.cloneFamily(cell))}")
      println(s"Qualifier:${Bytes.toString(CellUtil.cloneQualifier(cell))}")
      println(s"Value:${Bytes.toString(CellUtil.cloneValue(cell))}")
      println(s"TimeStamp:${cell.getTimestamp}")
      println("=======================================")
    }

    table.close()
    conn.close()
  }

  /**
   * 删除指定row数据
   *
   * @param conf
   * @param tableName
   * @param RowKey
   */
  def deleteRows(conf: Configuration, tableName: String, RowKey: String*) = {
    val conn: Connection = ConnectionFactory.createConnection(conf)
    val table: Table = conn.getTable(TableName.valueOf(tableName))
    val list: util.ArrayList[Delete] = new util.ArrayList[Delete]()
    for (row <- RowKey) {
      val delete: Delete = new Delete(Bytes.toBytes(row))
      list.add(delete)
    }
    table.delete(list)
    table.close()
    conn.close()
  }
}
